package com.liusy.taskengine;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.log4j.Logger;

import test.taskengine.SqlExecuteModel;

import com.liusy.taskengine.executemodel.DBExecuteModel;
import com.liusy.taskengine.listeners.AbstractCacheTaskCountListener;
import com.liusy.taskengine.listeners.AbstractCacheTaskFinishListener;
import com.liusy.taskengine.listeners.AbstractStartCacheTaskListener;
import com.liusy.taskengine.listeners.AbstractTaskReleaseFinishListener;
import com.liusy.taskengine.listeners.IReleaseTaskListener;
import com.liusy.taskengine.taskexecute.AbstractTaskExecute;
import com.liusy.taskengine.taskexecute.TaskExecute;
import com.liusy.taskengine.taskmodel.ITaskModel;

/**
 * 任务发布者
 * 
 * @author liusy 2012-12-11
 */
public class ReleaseTask extends Thread{
	
	private static Logger log = Logger.getLogger(SqlExecuteModel.class);
	
	private String releaseTaskName = "我 是任务发布者,正在发布任务....";
	
	
	
	//坐位   设置监听缓存任务多少时通知
	public final static String CACHE_TASK_COUNT_LISTENER = "CacheTaskCountListener";
	
	//坐位  监听所有任务完成，正在自我毁灭时通知
	public final static String CACHE_TASK_FINISH_LISTENER = "CacheTaskFinishListener";
	
	//坐位    用户的所有任务发布完时通知
	public final static String TASK_RELEASE_FINISH_LISTENER = "TaskReleaseFinishListener";
	
	//坐位   开始发布缓存任务时通知
	public final static String START_CACHE_TASK_LISTENER = "StartCacheTaskListener";
	
	
	
	/**
	 * 等待执行的任务 （缓存任务）
	 */
	private Vector<ITaskModel> vectorTask = new Vector<ITaskModel>();
	
	
	/**
	 * 监听者列表
	 */
	private ConcurrentMap<String,IReleaseTaskListener> mapListener = new ConcurrentHashMap<String,IReleaseTaskListener>();

	
	/**
	 * 允许的最大的执行器(工具 )
	 */
	private int maxThread = TaskEngineConstant.TASK_EXECUTE_THREAD_MAX;
	
	/**
	 * 执行者的老大(管理者)
	 */
	private TaskManager taskManager = null;
	
	/**
	 * 是否开始执行缓存任务
	 */
	private boolean cacheTaskStartExecute = false;
	
	
	
	/**
	 * 总共完成了的任务数量
	 * 
	 */
	private long taskNum = 0;
	
	
	public ReleaseTask(TaskManager taskManager)
	{
		this.taskManager=taskManager;
	}
	
	/**
	 * 发布任务 
	 * 
	 */
	public void ExeReleaseTask(ITaskModel taskModel){
		AbstractTaskExecute taskExecute =null;
	   	if(taskManager.countObservers()>0)
	   	{
	   		++taskNum;
	   		
	   		taskExecute = (AbstractTaskExecute) taskManager.getTaskExecute();
   		}
	   	if(null!=taskExecute)
	   		taskExecute.ExecuteTaskJob(taskModel);
	   	else
	   	{
	   		if(taskManager.threadCount>maxThread)
	   		{
	   			vectorTask.add(taskModel);
	   		}
	   		else{
	   			++taskNum;
		   		++taskManager.threadCount;
		   		taskExecute = new TaskExecute(taskModel, new DBExecuteModel(),taskManager);
		   		taskExecute.start();
	   		}
	   	}
	}
	
	/**
	 * 1、如有缓存任务，则自动发布。
	 * 2、如所有执行者完成任务，则通知所有的执行者自动销毁，并将自己(任务发布器)销毁
	 * 
	 */
	@Override
	public void run() {
		int count = 0;
		
		ExecuteStartCacheTaskListener();
		
		while(null!=taskManager)
		{
			if(cacheTaskStartExecute)
			{
				
				
				if(vectorTask.size()==0&&taskManager.countObservers()==taskManager.threadCount)
				{
					
					count +=taskManager.countObservers();
					log.debug("["+taskManager.countObservers()+"]位执行者所有任务已经完成！正在自我毁灭中！！其中["+(taskManager.threadCount-count)+"]位执行者目前尚位完成任务 !");
					taskManager.setRunFlag(0);
					taskManager.notifyObservers();
					if((taskManager.threadCount-count)<=0)
					{
						
						ExecuteCacheTaskFinishListener();
						
						ExecuteTaskReleaseFinishListener();
						
						log.debug("**************************************************************************************");
						log.debug("**************此次发布任务已经全部执行完成，感谢["+taskManager.threadCount+"]个执行器。**************");
						log.debug("**************************************************************************************");
						taskManager=null;
						this.interrupt();
					}
				}
				
				
				if(vectorTask.size()>0&&taskManager.countObservers()>0)
				{
					
					ExecuteCacheTaskCountListener();
					
					
					log.debug("正在发布缓存任务，还存在：["+vectorTask.size()+"]个缓存任务未发布,等待执行者的个数："+taskManager.countObservers());
					int flag = taskManager.countObservers()>=vectorTask.size()?vectorTask.size():taskManager.countObservers();
					
					for (int ii=0;ii<flag;ii++) {
						ITaskModel taskModel = vectorTask.firstElement();
							vectorTask.remove(taskModel);
							ExeReleaseTask(taskModel);
					}
				}
				try {
					this.sleep(TaskEngineConstant.RELEASE_TASK_RATE);
				} catch (InterruptedException e) {
					log.error(e.getMessage()+"ReleaseTask 任务发布器正在休息，不能 被打断 ！！");
				}
			}
		}
	}

	public String getReleaseTaskName() {
		return releaseTaskName;
	}

	public void setReleaseTaskName(String releaseTaskName) {
		this.releaseTaskName = releaseTaskName;
	}

	/**
	 * 添加监听者
	 * 
	 * @param listenerType  坐位
	 * @param lisener 监听者
	 */
	public void addListener(String listenerType,IReleaseTaskListener lisener){
		if(mapListener.containsKey(listenerType))
			mapListener.replace(listenerType, lisener);
		else
			mapListener.put(listenerType, lisener);
	}
	
	/**
	 * 删除监听者
	 * 
	 * @param listenerType 坐位
	 */
	public void removeListener(String listenerType){

		mapListener.remove(listenerType);
	}
	
	
	/**
	 * 得到当前剩余的缓存任务数量
	 * 
	 */
	public int getCacheTaskCount(){
		return vectorTask.size();
	}
	/**
	 *  是否开始执行缓存任务
	 * 
	 */
	public boolean isCacheTaskStartExecute(){
		return cacheTaskStartExecute;
	}
	
	
	/**
	 * 开始执行
	 * 
	 * 开始执行缓存中的任务，同时也决定执行器的生存。即（任务引引擎的生存）如果缓存中的任务执行完成后，任务引引擎将与执行器自我毁灭
	 * 
	 */
	public void StartCacheTaskStartExecute(){
		this.cacheTaskStartExecute = true;
	}
	
	/**
	 * ，暂停
	 * 
	 * 	只要保证缓存中还有任务，暂停操作才有效。否则任务引擎将自己我毁灭   
	 * 
	 */
	public void SuspendCacheTaskStartExecute(){
		this.cacheTaskStartExecute = false;
	}
	
	
	/**
	 * 得到共完成的任务数量
	 * 
	 * @return
	 */
	public long getTaskNum() {
		return taskNum;
	}

	/**
	 * 执行  开始发布缓存任务时通知
	 * 
	 * @param listenerType 坐位
	 */
	public void ExecuteStartCacheTaskListener(){
		
		if(this.mapListener.size()>0 && mapListener.containsKey(ReleaseTask.START_CACHE_TASK_LISTENER))
		{
			IReleaseTaskListener startCacheTaskListener = mapListener.get(ReleaseTask.START_CACHE_TASK_LISTENER);
			
			if(null != startCacheTaskListener && startCacheTaskListener instanceof AbstractStartCacheTaskListener)
			{
				((AbstractStartCacheTaskListener)startCacheTaskListener).StartCacheTaskListener(this);
			}
		}

	}
	
	
	
	/**
	 * 执行  用户的所有任务发布完时通知
	 * 
	 * @param listenerType 坐位
	 */
	public void ExecuteTaskReleaseFinishListener(){
		
		if(this.mapListener.size()>0 && mapListener.containsKey(ReleaseTask.TASK_RELEASE_FINISH_LISTENER))
		{
			IReleaseTaskListener taskReleaseFinishListener = mapListener.get(ReleaseTask.TASK_RELEASE_FINISH_LISTENER);
			
			if(null != taskReleaseFinishListener && taskReleaseFinishListener instanceof AbstractTaskReleaseFinishListener)
			{
				((AbstractTaskReleaseFinishListener)taskReleaseFinishListener).TaskReleaseFinishListener(this);
			}
		}

	}
	
	
	
	/**
	 * 执行  监听所有任务完成，正在自我毁灭时通知
	 * 
	 * @param listenerType 坐位
	 */
	public void ExecuteCacheTaskFinishListener(){
		
		if(this.mapListener.size()>0 && mapListener.containsKey(ReleaseTask.CACHE_TASK_FINISH_LISTENER))
		{
			IReleaseTaskListener cacheTaskFinishListener = mapListener.get(ReleaseTask.CACHE_TASK_FINISH_LISTENER);
			
			if(null != cacheTaskFinishListener && cacheTaskFinishListener instanceof AbstractCacheTaskFinishListener)
			{
				((AbstractCacheTaskFinishListener)cacheTaskFinishListener).CacheTaskFinishListener(this);
			}
		}
	}
	
	
	/**
	 * 执行 设置监听缓存任务多少时通知
	 * 
	 * @param listenerType 坐位
	 */
	public void ExecuteCacheTaskCountListener(){

		if(this.mapListener.size()>0 && mapListener.containsKey(ReleaseTask.CACHE_TASK_COUNT_LISTENER))
		{
			IReleaseTaskListener releaseTaskListener = mapListener.get(ReleaseTask.CACHE_TASK_COUNT_LISTENER);
			
			if(null != releaseTaskListener && releaseTaskListener instanceof AbstractCacheTaskCountListener)
			{
				int cacheTaskCount = ((AbstractCacheTaskCountListener)releaseTaskListener).getCacheTaskCount();
				
				if(cacheTaskCount>0 && vectorTask.size()<=cacheTaskCount && ((AbstractCacheTaskCountListener)releaseTaskListener).isPermissionExecute())
				{
						 ((AbstractCacheTaskCountListener)releaseTaskListener).CacheTaskCountListener(this);
				}
			}
		}
	}

	
}
